package cn.oowoo.rabbitmq.delay;

import cn.oowoo.rabbitmq.config.RabbitConfig;
import cn.oowoo.rabbitmq.config.RabbitExchangeTypeEnum;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
/**
* @Author: Zhangbin
* @Date: 2021/5/9 23:41
* @Description:  （TTL+DXL 实现延迟队列），绑定死信交换机，将过期的消息路由到死信队列，DealyConsumer端监听“死信”消息实现延迟队列
**/
public class DealyConsumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitConfig.createConnection();
        // 创建消息通道
        Channel channel = connection.createChannel();

        //创建一个topic交换机
        Map<String, Object> exchangeMap = new HashMap<>();
        exchangeMap.put("x-delayed-type", RabbitExchangeTypeEnum.TOPIC.getExchangeType());
        channel.exchangeDeclare(RabbitConfig.EXCHANGE_DEALY_NAME, RabbitExchangeTypeEnum.DEALY.getExchangeType(),false,false,exchangeMap);
        //创建队列
        channel.queueDeclare(RabbitConfig.QUEUE_DELAY_NAME,false,false,false,null);
        //绑定
        channel.queueBind(RabbitConfig.QUEUE_DELAY_NAME,RabbitConfig.EXCHANGE_DEALY_NAME,RabbitConfig.ROUTING_DELAY_KEY);


        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println(LocalDateTime.now() + "Received message : " + msg);
            }
        };
        channel.basicConsume(RabbitConfig.QUEUE_DELAY_NAME,true,consumer);
    }
}
